package com.zx.notification.consumer.listener;

import com.zx.notification.common.constants.NotificationConstants;
import com.zx.notification.common.event.ConfigReloadEvent;
import com.zx.notification.consumer.config.ChannelConfigManager;
import com.zx.redis.block.service.RedisBlockService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.event.EventListener;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.HashSet;
import java.util.Set;

/**
 * 渠道配置变更监听器
 * 监听渠道配置变更事件，清理相关缓存
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class ChannelConfigChangeListener {

    private final RedisBlockService redisBlockService;
    private final ChannelConfigManager channelConfigManager;
    private final ActiveMqMsgListener activeMqMsgListener;
    private final KafkaMsgListener kafkaMsgListener;
    private final RabbitMqMsgListener rabbitMqMsgListener;
    private final RocketMqMsgListener rocketMqMsgListener;
    private final ApplicationEventPublisher eventPublisher;

    /**
     * 处理配置重载事件
     * 当收到渠道配置相关的配置变更事件时，清理渠道配置缓存
     *
     * @param event 配置重载事件
     */
    @EventListener
    public void handleConfigReloadEvent(ConfigReloadEvent event) {
        try {
            // 检查是否是渠道配置变更
            if ("mq_config".equals(event.getConfigType())) {
                log.info("收到渠道配置变更事件: {}", event.getMessage());

                // 如果是特定渠道变更，则只清理该渠道的缓存
                String message = event.getMessage();
                if (message != null && message.startsWith("channel_type:")) {
                    String channelType = message.substring("channel_type:".length());
                    String cacheKey = NotificationConstants.RedisCache.CHANNEL_CONFIG_CACHE_KEY_PREFIX + channelType;
                    redisBlockService.deleteObject(cacheKey);
                    // 清理本地缓存
                    channelConfigManager.clearChannelCache(channelType);
                    log.info("已清理渠道[{}]的缓存", channelType);
                } else {
                    // 如果是全局渠道配置变更，则清理所有渠道配置缓存
                    clearAllChannelConfigCache();

                    // 重新加载各MQ监听器的内存缓存
                    reloadAllMqListenerCache();
                }
            }
        } catch (Exception e) {
            log.error("处理渠道配置变更事件时发生错误", e);
        }
    }

    /**
     * 清空所有渠道配置缓存
     */
    @SuppressWarnings("unchecked")
    private void clearAllChannelConfigCache() {
        try {
            // 使用scan命令查找所有匹配的缓存键
            Set<String> keys = (Set<String>) redisBlockService.redisTemplate.execute((RedisCallback<Set<String>>) connection -> {
                Set<String> result = new HashSet<>();
                Cursor<byte[]> cursor = connection.scan(ScanOptions.scanOptions()
                        .match(NotificationConstants.RedisCache.CHANNEL_CONFIG_CACHE_KEY_PREFIX + "*")
                        .count(1000)
                        .build());

                while (cursor.hasNext()) {
                    result.add(new String(cursor.next()));
                }
                cursor.close();
                return result;
            });

            // 删除所有匹配的键
            if (!CollectionUtils.isEmpty(keys)) {
                redisBlockService.deleteObject(keys);
                log.info("清空了{}个渠道配置缓存", keys.size());
            }
        } catch (Exception e) {
            log.error("清空渠道配置缓存时发生错误", e);
        }
    }

    /**
     * 重新加载所有MQ监听器的内存缓存
     */
    private void reloadAllMqListenerCache() {
        try {
            log.info("开始重新加载所有MQ监听器的内存缓存");

            // 重新加载各MQ监听器的内存缓存
            activeMqMsgListener.reloadChannelConfigurations();
            kafkaMsgListener.reloadChannelConfigurations();
            rabbitMqMsgListener.reloadChannelConfigurations();
            rocketMqMsgListener.reloadChannelConfigurations();

            log.info("所有MQ监听器的内存缓存重新加载完成");
        } catch (Exception e) {
            log.error("重新加载MQ监听器内存缓存时发生错误", e);
        }
    }
}